package com.gmscosmo.janus.queue;

import com.alibaba.fastjson.JSONObject;
import com.gmscosmo.janus.sendMessage.JanusAsyncSender;
import com.gmscosmo.janus.sendMessage.IMSender;
import com.gmscosmo.janus.sendMessage.JanusRequestSender;
import com.gmscosmo.janus.sendMessage.WebSocketSender;
import com.gmscosmo.janus.util.JanusLogUtil;
import com.gmscosmo.janus.util.JanusUtil;
import com.gmscosmo.model.Participant;
import com.gmscosmo.model.Room;
import com.rabbitmq.client.BuiltinExchangeType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.Executor;

/**
 * @author stephen.qiu
 */
@Component
public class RabbitMqConsumer {
    @Autowired
    JanusLogUtil janusLogUtil;
    @Autowired
    RabbitMqUtil rabbitMqUtil;
    @Autowired
    JanusUtil janusUtil;
    @Autowired
    RabbitmqHttpRequestUtil rabbitmqHttpRequestUtil;
    @Autowired
    IMSender imSender;
    @Autowired
    WebSocketSender webSocketSender;
    @Resource(name = "ThreadPoolTaskExecutor")
    Executor executor;

    @PostConstruct
    public void startConsumer() {
        String queueName = rabbitmqHttpRequestUtil.getQueueName("from-janus").get(0);
        janusLogUtil.log(queueName + "消费端启动", this.getClass(), JanusLogUtil.INFO);
        try {
            rabbitMqUtil.pullMq(queueName, queueName, BuiltinExchangeType.DIRECT, (record, routing_Key, deliveryTag, body) -> {
                try {
                    executor.execute(() -> {
                        handleMessage(body, queueName, routing_Key, deliveryTag);
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void handleMessage(byte[] body, String queueName, String routing_Key, long deliveryTag) {
        try {
            String message = new String(body, StandardCharsets.UTF_8);
            JSONObject jsonObject = JSONObject.parseObject(message);
            if (jsonObject.getString("transaction") != null) {
                String transaction = jsonObject.getString("transaction");
                if (janusUtil.isTransaction(transaction)) {
                    janusLogUtil.log(queueName + "消费数据：" + message, this.getClass(), JanusLogUtil.DEBUG);
                    rabbitMqUtil.ack(routing_Key, deliveryTag);
                    JanusRequestSender.notifyObservers(transaction, jsonObject);
                    JanusAsyncSender.notifyObservers(transaction, jsonObject);
                } else {
                    janusLogUtil.log(queueName + "拒绝数据：" + message, this.getClass(), JanusLogUtil.DEBUG);
                    rabbitMqUtil.nack(routing_Key, deliveryTag);
                }
            } else {
                janusLogUtil.log(queueName + "事件：" + message, this.getClass(), JanusLogUtil.DEBUG);
                rabbitMqUtil.ack(routing_Key, deliveryTag);
                if (message.contains("leaving")) {
                    handleLeavingEvent(jsonObject);
                    return;
                }
                if (message.contains("started")) {
                    handleStartedEvent(jsonObject);
                }
                if (message.contains("timeout")) {
                    handleTimeoutEvent(jsonObject);
                }
                String userId = janusUtil.getUserIdBySession(jsonObject.getLong("session_id"));
                imSender.sendEventToListener(userId, jsonObject);
                webSocketSender.sendEventToListener(userId, jsonObject);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void handleTimeoutEvent(JSONObject jsonObject) {
        long session_id = jsonObject.getLong("session_id");
        janusUtil.handleTimeout(session_id);
    }

    private void handleLeavingEvent(JSONObject jsonObject) {
        JSONObject plugindata = jsonObject.getJSONObject("plugindata");
        JSONObject data = plugindata.getJSONObject("data");
        String roomId = String.valueOf(data.getInteger("room"));
        long publisher_id = data.getLong("leaving");
        Room room = janusUtil.getAVRoomByRoomId(roomId);
        List<Participant> participants = room.getParticipantList();
        for (Participant participant : participants) {
            if (participant.getPublisher_id() == publisher_id) {
                janusUtil.handleLeaving(participant.getSession_id());
                break;
            }
        }
    }

    private void handleStartedEvent(JSONObject jsonObject) {
        JSONObject plugindata = jsonObject.getJSONObject("plugindata");
        long session_id = jsonObject.getLong("session_id");
        long handle_id = jsonObject.getLong("sender");
        JSONObject data = plugindata.getJSONObject("data");
        String userId = janusUtil.getUserIdBySession(session_id);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        String targetUserId = participant.getTargetUserId(handle_id);
        if (targetUserId != null) {
            data.put("display", targetUserId);
            plugindata.put("data", data);
            jsonObject.put("plugindata", plugindata);
        }
    }
}
